/**
 * Copyright (C) 2010, 2011 Neofonie GmbH
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.analysis.restapi.controllers;

import eu.dicodeproject.analysis.restapi.SchemaConfig;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;

/**
 * @author jakob
 */
public class TweetIdsController extends FilteringController {

  private Map<String,SchemaConfig> schemaConfigMap;

  /**
   * Returns term vectors for documents in a time range and other filters.
   * @param source source table (e.g. twitter or blogs)
   * @param start start date (inclusive) for time range
   * @param stop stop date (exclusive) for time range
   * @param language language to filter by
   * @param regex regular expression that has to match parts of the document text
   * @param limit user-specific result limit; cannot be larger than maximumResultLimit
   * @param callback JSON callback string
   * @param response object to write the result to
   * @throws java.io.IOException inherited from HBase Scan and HttpServletResponse.write
   * @throws twitter4j.internal.org.json.JSONException inherited from JSONObject constructor
   */
  @RequestMapping(value = "{source}/tweetIds", method = RequestMethod.GET)
  public void vectors(@PathVariable String source,
      @RequestParam(required = false) String start, @RequestParam(required = false) String stop,
      @RequestParam(required = false) String language, @RequestParam(required = false) String regex,
      @RequestParam(required = false) Integer limit, @RequestParam(required = false) String callback,
      HttpServletResponse response) throws IOException, JSONException {

    // get schema
    SchemaConfig schemaConfig = schemaConfigMap.get(source);

    // collect parameters for result
    JSONObject params = new JSONObject();

    // set HBase scan
    Scan scan = getScan(schemaConfig, start, stop, language, regex, params);

    // get HBase table
    HTableInterface hTable = getHTable(schemaConfig.getTableName());

    // get result limit
    int resultLimit = getResultLimit(limit);
    params.put("limit", resultLimit);

    // read results from scan into a list of maps
    List<String> tweetList = new LinkedList<String>();
    int totalResults = 0;
    for (Result result : hTable.getScanner(scan)) {
      String rowId = Bytes.toString(result.getRow());
      String tweetId = rowId.substring(schemaConfig.idStartPositionInKey());  // skip creation date
      tweetList.add(tweetId);

      totalResults++;
      if (totalResults >= resultLimit) {
        break;
      }
    }
    params.put("results", totalResults);

    // put result list and parameters map into a JSON
    JSONObject json = new JSONObject();
    json.put("parameters", params);
    json.put("tweets", tweetList);

    // write out the result JSON
    respond(callback, json, response);
  }

  public void setSchemaConfigMap(Map<String, SchemaConfig> schemaConfigMap) {
    this.schemaConfigMap = schemaConfigMap;
  }

}
